package gbench.appdemo.jdbcapp.gsx;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * https://github.com/LMAX-Exchange/disruptor/
 * @author gbench
 *
 */
public class GlobalSecurityExchange {

    public static final class StubEvent {
        private int value;
        private String testString;
        public static final EventTranslatorTwoArg<StubEvent, Integer, String> TRANSLATOR =
            new EventTranslatorTwoArg<StubEvent, Integer, String>() {
            @Override
            public void translateTo(StubEvent event, long sequence, Integer arg0, String arg1) {
                event.setValue(arg0);
                event.setTestString(arg1);
            }
        };

        public StubEvent(int i) {
            this.value = i;
        }

        public void copy(StubEvent event) {
            value = event.value;
        }

        public int getValue() {
            return value;
        }

        public void setValue(int value) {
            this.value = value;
        }

        public String getTestString() {
            return testString;
        }

        public void setTestString(final String testString) {
            this.testString = testString;
        }

        public static final EventFactory<StubEvent> EVENT_FACTORY = new EventFactory<StubEvent>() {
            public StubEvent newInstance() {
                return new StubEvent(-1);
            }
        };

        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + value;
            return result;
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj) {
                return true;
            }
            if (obj == null) {
                return false;
            }
            if (getClass() != obj.getClass()) {
                return false;
            }
            StubEvent other = (StubEvent) obj;

            return value == other.value;
        }
    }

    private static class MessageProduce implements Runnable {
        Disruptor<StubEvent> disruptor;
        int start;
        int over;

        MessageProduce(Disruptor<StubEvent> disruptor, int start, int over) {
            this.disruptor = disruptor;
            this.start = start;
            this.over = over;
        }

        @Override
        public void run() {
            for (int i = start; i < over + start; i++) {
                RingBuffer<StubEvent> ringBuffer = disruptor.getRingBuffer();
                long sequence = ringBuffer.next();
                try {
                    StubEvent event = ringBuffer.get(sequence);
                    event.setTestString("msg => " + i);
                } finally {
                    ringBuffer.publish(sequence);
                }
            }
        }
    }

    private static class DynamicHandler implements WorkHandler<StubEvent>, LifecycleAware {
        private final CountDownLatch shutdownLatch = new CountDownLatch(1);

        @Override
        public void onStart() {

        }

        @Override
        public void onShutdown() {
            shutdownLatch.countDown();
        }

        void awaitShutdown() throws InterruptedException {
            shutdownLatch.await();
        }

        @Override
        public void onEvent(StubEvent event) throws Exception {
            System.out.println(event.getTestString() + " ,thread ==> " + Thread.currentThread().getId());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Sequence workSequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
        ExecutorService executor = Executors.newCachedThreadPool(DaemonThreadFactory.INSTANCE);

        // Build a disruptor and start it.
        Disruptor<StubEvent> disruptor = new Disruptor<>(StubEvent.EVENT_FACTORY, 1024, DaemonThreadFactory.INSTANCE);
        RingBuffer<StubEvent> ringBuffer = disruptor.start();

        // Construct 2 batch event processors.
        GlobalSecurityExchange.DynamicHandler handler1 = new GlobalSecurityExchange.DynamicHandler();
        WorkProcessor<StubEvent> processor1 = new WorkProcessor<>(ringBuffer, ringBuffer.newBarrier(), handler1,
                new FatalExceptionHandler(), workSequence);

        GlobalSecurityExchange.DynamicHandler handler2 = new GlobalSecurityExchange.DynamicHandler();
        WorkProcessor<StubEvent> processor2 = new WorkProcessor<>(ringBuffer, ringBuffer.newBarrier(), handler2,
                new FatalExceptionHandler(), workSequence);

        // Dynamically add both sequences to the ring buffer
        ringBuffer.addGatingSequences(processor1.getSequence());
        // Start the new batch processors.
        executor.execute(processor1);

        ringBuffer.addGatingSequences(processor2.getSequence());
        executor.execute(processor2);

        Thread thread1 = new Thread(new MessageProduce(disruptor, 0, 100));
        thread1.start();

        Thread.sleep(1000);

        // Remove a processor.
        // Stop the processor , processor2.haltLater() will wait for processor2 message
        // processing to complete
        processor2.halt();

        Thread thread2 = new Thread(new MessageProduce(disruptor, 100, 200));
        thread2.start();

        // Wait for shutdown the complete
        handler2.awaitShutdown();
        // Remove the gating sequence from the ring buffer
        ringBuffer.removeGatingSequence(processor2.getSequence());
    }

}
